package cn.edu.laud.sany.datatest;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.CfDef;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.KsDef;
import org.apache.cassandra.thrift.SchemaDisagreementException;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;

import cn.edu.laud.sany.types.ImportType;
import cn.edu.laud.sany.utils.Config;
import cn.edu.laud.sany.utils.MyLog;
import cn.edu.laud.sany.utils.ProcessManager;
import cn.edu.laud.sany.utils.TimeUtils;

public class LaUDImportTest
{
	// config params
	public static int thread = 1;
	public static int schema = 1;
	public static int laudPort = Config.InitLaudPort;
	public static String[] IP = new String[] {};
	public static String[] host = new String[] {};
	public static String keyspace = "";
	public static ImportType importType;
	public static String csvFolder = "";
	public static String cfsFile = "";
	public static String ignoreParaIdsFile = "";
	public static String cQLTranscation = "ANY";
	public static int replicationFactor = 2;
	public static boolean errorDataStop = false;
	public static boolean ignoreSomeParaId = false;
	
	// used constants
	public static Integer[] cFS;
	public static HashSet<Integer> ignoreParaIds = new HashSet<Integer>();
	public static String cfNameFormat = "";
	public static String cfPrimaryKeyAlias = "";
	public static String cfComparatorType = Config.CFComparatorType0;
	public static String logFileFolder = "log";
	public static String logFileName = "mylogtest.txt";
	public static String cQLInsertFormat = "";
	public static int testCycles = 1;
	public static int testCycleData = 1000000;
	
	public static long nowCycleCount = 0;
	
	private static MyLog logger;
	
	/**
	 * @param args
	 * @throws ClassNotFoundException
	 * @throws SQLException
	 */
	public static void main(String[] args) throws ClassNotFoundException, SQLException, IOException
	{
		initMyLog();
		doMain(args);
	}
	
	public static String getLogFilePath()
	{
		return (logFileFolder + "/" + logFileName);
	}
	
	protected static void initMyLog()
	{
		File logFolderFile = new File(logFileFolder);
		if (logFolderFile.exists() == false)
		{
			logFolderFile.mkdir();
		}
		logger = MyLog.getMyLog(getLogFilePath(), true);
	}
	
	protected static void doMain(String[] args) throws ClassNotFoundException, SQLException,
			IOException
	{
		if (args.length % 2 == 1 || args.length < 1)
		{
			System.err.println(Config.CommandHelpString);
			return;
		}
		
		try
		{
			parseParams(args);
		}
		catch (Exception e)
		{
			System.err.println(Config.CommandHelpString);
			return;
		}
		
		// test csv and cfs files
		if (testCsvFoler() == false)
		{
			System.err.println("Can not find csv folder!");
			return;
		}
		if (testCfFile() == false)
		{
			System.err.println("Can not find CFs file!");
			return;
		}
		if (getCFNames() == false)
		{
			System.err.println("Can not read CFs correctly!");
			return;
		}
		if (ignoreSomeParaId && getIgnoreParaIds() == false)
		{
			System.err.println("Can not read ignoreParaIds correctly!");
			return;
		}
		
		long starttime = System.nanoTime();
		logger.info("import test start, time:" + starttime);
		
		try
		{
			Class.forName("org.apache.cassandra.cql.jdbc.CassandraDriver");
		}
		catch (ClassNotFoundException e)
		{
			e.printStackTrace();
			return;
		}
		catch (Exception e)
		{
			e.printStackTrace();
			return;
		}
		
		// set column family
		switch (schema)
		{
			case 1:
				// 工况表
				cfNameFormat = Config.CFNameFormat1;
				cfPrimaryKeyAlias = Config.CFPrimaryKeyAlias1;
				cQLInsertFormat = Config.CQLInsertFormat1;
				break;
			case 2:
				// 起重机表
				cfNameFormat = Config.CFNameFormat2;
				cfPrimaryKeyAlias = Config.CFPrimaryKeyAlias2;
				cQLInsertFormat = Config.CQLInsertFormat2;
				break;
			default:
				// 其他表
				cfNameFormat = Config.CFNameFormat0;
				cfPrimaryKeyAlias = Config.CFPrimaryKeyAlias0;
				cQLInsertFormat = Config.CQLInsertFormat0;
				break;
		}
		
		if (importType == ImportType.CLEAR_DATA)
		{
			if (dropTableWithThrift(keyspace) == false)
			{
				logger.info("Drop keyspace failed!");
				return;
			}
			return;
		}
		if (importType == ImportType.NEW_KEYSPACE)
		{
			if (createTable(schema, keyspace) == false)
			{
				logger.info("Create keyspace failed!");
				return;
			}
		}
		if (importType == ImportType.RECREATE_KEYSPACE)
		{
			if (dropTableWithThrift(keyspace) == false)
			{
				logger.info("Drop keyspace failed!");
			}
			if (createTable(schema, keyspace) == false)
			{
				logger.info("Create keyspace failed!");
				return;
			}
		}
		
		try
		{
			ProcessManager.InitProcessManager(logger, csvFolder, logFileFolder);
			ProcessManager.writeProgressSpeedlog = false;
			ProcessManager.updateTimesToWriteFile = 2;
		}
		catch (Exception e)
		{
			System.err.println("Init ProcessManager failed: " + e.getMessage());
			return;
		}
		
		LaUDTest_JDBC.sqlQueue = new ConcurrentLinkedQueue<String>();
		
		LaUDTest_JDBC[] testers = new LaUDTest_JDBC[thread];
		LaUDTest_CSVReader[] testReaders = new LaUDTest_CSVReader[thread];
		for (int i = 0; i < thread; i++)
		{
			String jdbcString = host[i % host.length];
			LaUDTest_JDBC tester = new LaUDTest_JDBC(i, jdbcString, keyspace, getLogFilePath(), 0);
			testers[i] = tester;
			tester.begin_Test_Import();
			LaUDTest_CSVReader reader = new LaUDTest_CSVReader(i);
			testReaders[i] = reader;
		}
		
		int allImportCount = 0;
		for (int k = 0; k < testCycles; k++)
		{
			nowCycleCount = 0;
			ProcessManager.requestUpdateQueue();
			int availableFileCount = ProcessManager.getSourceCount();
			int readThread = thread;
			if (availableFileCount == 0)
			{
				break;
			}
			if (availableFileCount < thread)
			{
				readThread = availableFileCount;
			}
			long oneTestStartTime = System.nanoTime();
			logger.info("Data import test cycle [" + k + "] begin...");
			
			// read
			BlockingQueue<Runnable> queue0 = new LinkedBlockingQueue<Runnable>();
			ThreadPoolExecutor executor0 = new ThreadPoolExecutor(readThread, readThread, 100,
					TimeUnit.DAYS, queue0);
			for (int i = 0; i < readThread; i++)
			{
				executor0.execute(testReaders[i]);
			}
			executor0.shutdown();
			try
			{
				executor0.awaitTermination(100, TimeUnit.DAYS);
			}
			catch (InterruptedException e1)
			{
				logger.error("Main thread interrupted while waiting " + e1.getMessage());
			}
			logger.info("Build sql batch finished! count = " + LaUDTest_JDBC.sqlQueue.size());
			
			// import
			long oneTestStartTime2 = System.nanoTime();
			BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
			ThreadPoolExecutor executor = new ThreadPoolExecutor(thread, thread, 100,
					TimeUnit.DAYS, queue);
			for (int i = 0; i < thread; i++)
			{
				executor.execute(testers[i]);
			}
			executor.shutdown();
			try
			{
				executor.awaitTermination(100, TimeUnit.DAYS);
			}
			catch (InterruptedException e1)
			{
				logger.error("main thread interrupted while waiting " + e1.getMessage());
			}
			long oneTestEndTime = System.nanoTime();
			long readTime = oneTestStartTime2 - oneTestStartTime;
			long importTime = oneTestEndTime - oneTestStartTime2;
			long speed = 1000000000L * nowCycleCount / importTime;
			ProcessManager.processTestLog(nowCycleCount, readTime, importTime);
			allImportCount += nowCycleCount;
			logger.info("Cycle[" + k + "] test finish. RealImport:" + allImportCount
					+ " readCsvTime:" + TimeUtils.nanoSecondToString(readTime) + " ImportTime:"
					+ TimeUtils.nanoSecondToString(importTime) + " ImportSpeed:" + speed);
		}
		ProcessManager.applicationExit = true;
		long endTime = System.nanoTime();
		logger.info("Import test program's total run time:"
				+ TimeUtils.nanoSecondToString((endTime - starttime)));
	}
	
	protected static void parseParams(String[] args)
	{
		for (int i = 0; i < args.length; i += 2)
		{
			if (args[i].equals("--thread"))
			{
				thread = Integer.valueOf(args[i + 1]);
			}
			else if (args[i].equals("--schema"))
			{
				schema = Integer.valueOf(args[i + 1]);
				if (keyspace == null)
				{
					keyspace = "laud_sany_case" + schema;
				}
			}
			else if (args[i].equals("--type"))
			{
				String valString = args[i + 1];
				if (valString.equals("new"))
				{
					importType = ImportType.NEW_KEYSPACE;
				}
				else if (valString.equals("append"))
				{
					importType = ImportType.APPEND_DATA;
				}
				else if (valString.equals("clear"))
				{
					importType = ImportType.CLEAR_DATA;
				}
				else
				{
					importType = ImportType.RECREATE_KEYSPACE;
				}
			}
			else if (args[i].equals("--laudport"))
			{
				laudPort = Integer.valueOf(args[i + 1]);
			}
			else if (args[i].equals("--laudip"))
			{
				IP = args[i + 1].split(",");
				host = new String[IP.length];
				for (int k = 0; k < IP.length; k++)
				{
					host[k] = String.format(Config.JDBCFormat, IP[k], laudPort);
					logger.debug(host[k]);
				}
			}
			else if (args[i].equals("--replica"))
			{
				replicationFactor = Integer.valueOf(args[i + 1]);
			}
			else if (args[i].equals("--keyspace"))
			{
				keyspace = args[i + 1];
			}
			else if (args[i].equals("--csvpath"))
			{
				csvFolder = args[i + 1];
			}
			else if (args[i].equals("--cfsfile"))
			{
				cfsFile = args[i + 1];
			}
			else if (args[i].equals("--ignorefile"))
			{
				ignoreSomeParaId = true;
				ignoreParaIdsFile = args[i + 1];
			}
			else if (args[i].equals("--errorStop"))
			{
				if (args[i + 1].equals("1"))
				{
					errorDataStop = true;
				}
			}
			else if (args[i].equals("--comparator"))
			{
				cfComparatorType = Config.CFComparatorTypePrefix + args[i + 1];
			}
			else if (args[i].equals("--transcation"))
			{
				cQLTranscation = args[i + 1];
			}
			else if (args[i].equals("--testcycle"))
			{
				testCycles = Integer.valueOf(args[i + 1]);
				if (testCycles < 1)
				{
					testCycles = 1;
				}
			}
			else if (args[i].equals("--cycledata"))
			{
				testCycleData = Integer.valueOf(args[i + 1]);
			}
		}
		logger.info("thread = " + thread);
		logger.info("schema = " + schema);
		logger.info("type = " + importType);
		logger.info("laud port = " + laudPort);
		String ipsString = "";
		for (String ip : IP)
		{
			ipsString += ip + "; ";
		}
		logger.info("laud ip = " + ipsString);
		logger.info("replication factor = " + replicationFactor);
		logger.info("keyspace = " + keyspace);
		logger.info("csv path = " + csvFolder);
		logger.info("cfs file = " + cfsFile);
		logger.info("ignore file = " + ignoreParaIdsFile);
		logger.info("errorData Stop = " + errorDataStop);
		logger.info("comparator = " + cfComparatorType);
		logger.info("transcation = " + cQLTranscation);
		logger.info("testcycle = " + testCycles);
		logger.info("cycledata = " + testCycleData);
		
	}
	
	protected static boolean getCFNames()
	{
		boolean flag = true;
		File file = new File(cfsFile);
		BufferedReader bReader = null;
		String paraIdString = null;
		ArrayList<Integer> cfList = new ArrayList<Integer>();
		try
		{
			bReader = new BufferedReader(new FileReader(file));
			while ((paraIdString = bReader.readLine()) != null)
			{
				paraIdString = paraIdString.trim();
				if (paraIdString.equals(""))
				{
					continue;
				}
				int paraId = Integer.valueOf(paraIdString);
				cfList.add(paraId);
			}
			
		}
		catch (IOException e1)
		{
			flag = false;
			logger.error("Can not open cfs file");
		}
		catch (NumberFormatException e2)
		{
			flag = false;
			logger.error("Cfs file contains invalid integer");
		}
		finally
		{
			if (flag == false)
			{
				logger.info("Get cfs failed!");
			}
			try
			{
				if (bReader != null)
				{
					bReader.close();
				}
			}
			catch (IOException e)
			{
				logger.info(e.getMessage());
			}
		}
		if (flag == true)
		{
			cFS = cfList.toArray(new Integer[0]);
		}
		return flag;
	}
	
	protected static boolean testCsvFoler()
	{
		try
		{
			File file = new File(csvFolder);
			if (file.exists())
			{
				return true;
			}
		}
		catch (Exception e)
		{
			e.printStackTrace();
		}
		return false;
	}
	
	protected static boolean testCfFile()
	{
		try
		{
			File file = new File(cfsFile);
			if (file.exists())
			{
				return true;
			}
		}
		catch (Exception e)
		{
			e.printStackTrace();
		}
		return false;
	}
	
	protected static boolean getIgnoreParaIds()
	{
		ignoreParaIds = new HashSet<Integer>();
		boolean flag = true;
		File file = new File(ignoreParaIdsFile);
		BufferedReader bReader = null;
		String paraIdString = null;
		try
		{
			bReader = new BufferedReader(new FileReader(file));
			while ((paraIdString = bReader.readLine()) != null)
			{
				paraIdString = paraIdString.trim();
				if (paraIdString.equals(""))
				{
					continue;
				}
				int paraId = Integer.valueOf(paraIdString);
				ignoreParaIds.add(paraId);
			}
			
		}
		catch (IOException e1)
		{
			flag = false;
			logger.error("Can not open ignoreParaIds file!");
		}
		catch (NumberFormatException e2)
		{
			flag = false;
			logger.error("IgnoreParaIds file contains invalid integer!");
		}
		finally
		{
			if (flag == false)
			{
				logger.info("Get ignoreParaIds failed!");
			}
			try
			{
				if (bReader != null)
				{
					bReader.close();
				}
			}
			catch (IOException e)
			{
				logger.info(e.getMessage());
			}
		}
		return flag;
	}
	
	protected static boolean dropTableWithThrift(String keyspace) throws IOException
	{
		boolean flag = true;
		TTransport tr = new TFramedTransport(new TSocket(IP[0], laudPort));
		TProtocol proto = new TBinaryProtocol(tr);
		Cassandra.Client client = new Cassandra.Client(proto);
		try
		{
			tr.open();
			try
			{
				logger.info("Waiting server to drop keyspace...");
				client.system_drop_keyspace(keyspace);
				logger.info("Drop keyspace successful!");
			}
			catch (InvalidRequestException e1)
			{
				flag = false;
				logger.error(e1.toString());
			}
			catch (SchemaDisagreementException e1)
			{
				flag = false;
				logger.error(e1.toString());
			}
			catch (TException e1)
			{
				flag = false;
				logger.error(e1.toString());
			}
			
		}
		catch (TTransportException e2)
		{
			flag = false;
			logger.error(e2.toString());
			logger.error("Can not connect LaUD server, IP:" + IP[0] + " Port:" + laudPort);
		}
		finally
		{
			tr.close();
		}
		return flag;
	}
	
	protected static boolean createTableWithThrift(String keyspace)
	{
		boolean flag = true;
		TTransport tr = new TFramedTransport(new TSocket(IP[0], laudPort));
		TProtocol proto = new TBinaryProtocol(tr);
		Cassandra.Client client = new Cassandra.Client(proto);
		List<CfDef> cfDefs = new ArrayList<CfDef>();
		try
		{
			tr.open();
			for (int paraId : cFS)
			{
				CfDef cfDef = new CfDef(keyspace, String.format(cfNameFormat,
						String.valueOf(paraId)));
				cfDef.setKey_alias(cfPrimaryKeyAlias.getBytes());
				cfDef.setComparator_type(cfComparatorType);
				cfDef.setDefault_validation_class(Config.CFCalidationClass);
				cfDef.setKey_validation_class(Config.CFCalidationClass);
				cfDefs.add(cfDef);
			}
			KsDef ks_def = new KsDef(keyspace, Config.KSStrategy, cfDefs);
			ks_def.setStrategy_options(Collections.singletonMap("replication_factor",
					String.valueOf(replicationFactor)));
			try
			{
				logger.info("Waiting server to create keyspace and schema...");
				client.system_add_keyspace(ks_def);
				logger.info("Create keyspace and schema successful!");
			}
			catch (InvalidRequestException e)
			{
				flag = false;
				logger.error(e.toString());
			}
			catch (SchemaDisagreementException e)
			{
				flag = false;
				logger.error(e.toString());
			}
			catch (TException e)
			{
				flag = false;
				logger.error(e.toString());
			}
		}
		catch (TTransportException e2)
		{
			flag = false;
			logger.error(e2.toString());
			logger.error("Can not connect LaUD server, IP:" + IP[0] + " Port:" + laudPort);
		}
		finally
		{
			tr.close();
		}
		return flag;
	}
	
	protected static boolean createTable(int casetype, String keyspace) throws IOException
	{
		return createTableWithThrift(keyspace);
	}
	
}
